﻿using System;
using System.Collections;
using Microsoft.SqlServer.Dts.Pipeline;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using System.Threading.Tasks;
using System.Collections.Generic;

namespace com.webkingsoft.ssis.pipelinecomponent
{
    [DtsPipelineComponent
        (
            DisplayName = "WS MassTrimmer",
            ComponentType = ComponentType.Transform,
            IconResource = "com.webkingsoft.ssis.pipelinecomponent.MassTrimmer.ico"
        )
    ]
    public class MassTrimmer : PipelineComponent
    {
        public const int PARALLEL_THD = 4;
        ArrayList indexesToTrim = new ArrayList();
        ArrayList indexesToCopy = new ArrayList();
        object[] indexesToTrimArray;
        bool userParallelism = false;

        public override void ProvideComponentProperties()
        {
            base.ProvideComponentProperties();
            ComponentMetaData.InputCollection[0].Name = "Input to trim";
            ComponentMetaData.OutputCollection[0].Name = "Trimmed outputs";
            ComponentMetaData.Description = "Webkingsoft MassTrimmer: allows you to trim all columns of a table in a couple of clicks.";
            ComponentMetaData.ContactInfo = "Alberto Geniola, albertogeniola@gmail.com";
        }

        /// <summary>
        /// This method is called once before execution. We set up this component by choosing which strategy to use when processisng input:
        /// If we have less than PARALLEL_THD columns as inputs, we'll use a serial approach. Otherwise we'll use parallelism.
        /// </summary>
        public override void PreExecute()
        {
            IDTSInput100 input = ComponentMetaData.InputCollection[0];
            IDTSInputColumnCollection100 inputColumns = input.InputColumnCollection;

            foreach (IDTSInputColumn100 column in inputColumns)
            {
                if (column.DataType == DataType.DT_STR || column.DataType == DataType.DT_WSTR)
                {
                    indexesToTrim.Add((int)BufferManager.FindColumnByLineageID(input.Buffer, column.LineageID));
                }
                else
                    indexesToCopy.Add((int)BufferManager.FindColumnByLineageID(input.Buffer, column.LineageID));
            }

            if (indexesToTrim.Count > PARALLEL_THD)
                userParallelism = true;
            else
                userParallelism = false;

            indexesToTrimArray = indexesToTrim.ToArray();
        }

        /// <summary>
        /// This method will process each row at once. The .NET framework uses to collect some rows in a buffer and pass the whole buffer to the
        /// function. So we have to process multiple rows at once.
        /// </summary>
        /// <param name="inputID"></param>
        /// <param name="buffer"></param>
        public override void ProcessInput(int inputID, PipelineBuffer buffer)
        {
            while (buffer.NextRow())
            {
                // Considering this component is usefoul when the input contains many columns, it's crucial to 
                // optimize parallelism. If number of columns is lower than PARALLEL_THD, we will use a serial cycle, otherwise we 
                // use a parallel cycle.
                if (userParallelism)
                    Parallel.ForEach(indexesToTrimArray, new ParallelOptions { MaxDegreeOfParallelism = PARALLEL_THD }, columnIndex =>
                    {
                        string str = buffer.GetString((int)columnIndex);
                        buffer.SetString((int)columnIndex, str.Trim());
                    });
                else
                    foreach (int columnIndex in indexesToTrim)
                    {
                        string str = buffer.GetString((int)columnIndex);
                        buffer.SetString((int)columnIndex, str.Trim());
                    }
            }
        }
    }
}
